package com.ffcs.crmd.tsp.core.netty;

import java.util.HashMap;
import java.util.Iterator;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;

import com.ffcs.crmd.tsp.api.dto.netty.RemotingCommand;
import com.ffcs.crmd.tsp.api.netty.INettyInvokeCallback;
import com.ffcs.crmd.tsp.api.netty.INettyProcessor;
import com.ffcs.crmd.tsp.common.exception.TspRemotingException;
import com.ffcs.crmd.tsp.common.netty.RequestCode;
import com.ffcs.crmd.tsp.common.netty.ResponseCode;
import com.ffcs.crmd.tsp.common.netty.enu.NettyExceptionType;
import com.ffcs.crmd.tsp.core.util.RemotingUtil;

import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;

/**
 * 
 * 功能说明:netty通信共用抽象父类
 *
 * @author ZHONGFUHUA
 * 
 * @since 1.0.0-SNAPSHOT
 *
 */
public abstract class NettyRemotingAbstract {
    
    /**
     * 信号量，异步调用情况会使用，防止本地Netty缓存请求过多
     */
    private final Semaphore                                                   semaphoreAsync;
    
    /**
     * 信号量，Oneway情况会使用，防止本地Netty缓存请求过多
     */
    private final Semaphore                                                   semaphoreOneway;
    
    /**
     * 缓存所有对外请求 Integer(opaque)
     */
    private final ConcurrentHashMap<Integer, ResponseFuture>                  responseTable  = new ConcurrentHashMap<Integer, ResponseFuture>();
    
    /**
     * 注册的各个RPC处理器(远程过程调用协议)Integer(request code)
     */
    private final HashMap<Integer, RpcPair<INettyProcessor, ExecutorService>> processorTable = new HashMap<Integer, RpcPair<INettyProcessor, ExecutorService>>();
    
    public NettyRemotingAbstract(int permitsOneway, int permitsAsync) {
        this.semaphoreAsync = new Semaphore(permitsAsync, true);
        this.semaphoreOneway = new Semaphore(permitsOneway, true);
    }
    
    /**
     * 
     * 功能说明:处理远程终端发送过来的命令
     * 
     * @param ctx 上下文
     * @param remotingCommand 远程命令
     */
    public void processMessageReceived(ChannelHandlerContext ctx, RemotingCommand remotingCommand) {
        final RemotingCommand cmd = remotingCommand;
        if (cmd != null) {
            switch (cmd.getType()) {
                case REQUEST_COMMAND:
                    processRequestCommand(ctx, cmd);
                    break;
                case RESPONSE_COMMAND:
                    processResponseCommand(ctx, cmd);
                    break;
            }
        }
    }
    
    /**
     * 
     * 功能说明:处理远程终端的请求
     * 
     * @param ctx 上下文
     * @param cmd 返回结果
     */
    public void processRequestCommand(final ChannelHandlerContext ctx, final RemotingCommand cmd) {
        final RpcPair<INettyProcessor, ExecutorService> rpcpair = processorTable.get(cmd.getCode());
        getLog().debug("NETTY-COMMON: recive a request opaque[{}] code[{}] from channel=[{}]", cmd.getOpaque(),
            RequestCode.getName(cmd.getCode()), ctx.channel());
        if (rpcpair != null) {
            Runnable run = new Runnable() {
                @Override
                public void run() {
                    try {
                        RemotingCommand response = rpcpair.getExecuteBussObj().processRequest(ctx, cmd);
                        if (!cmd.isOneway()) {
                            if (response != null) {
                                response.setOpaque(cmd.getOpaque());
                                ctx.writeAndFlush(response);
                            } else {
                                getLog().warn(
                                    "NETTY-COMMON: process a is't one way type request opaque[{}] code[{}] from channel[{}] but return a null reponse.",
                                    ctx.channel(), cmd.getOpaque(),
                                    RequestCode.getName(cmd.getCode()));
                            }
                        }
                    } catch (Throwable e) {
                        getLog().error(
                            "NETTY-COMMON: process a request opaque[{}] code[{}] from channel[{}] exception:",
                            cmd.getOpaque(), RequestCode.getName(cmd.getCode()), ctx.channel(), e);
                    }
                }
            };
            rpcpair.getExecuteThreadPool().submit(run);
        } else {
            if (!cmd.isOneway()) {
                getLog().warn(
                    "NETTY-COMMON: recive a request opaque[{}] code[{}] from channel[{}] but not supported processor.",
                    cmd.getOpaque(), RequestCode.getName(cmd.getCode()), ctx.channel());
                final RemotingCommand response = RemotingCommand
                    .createResponseCommand(ResponseCode.REQUEST_CODE_NOT_SUPPORTED);
                response.setOpaque(cmd.getOpaque());
                ctx.writeAndFlush(response);
            }
        }
    }
    
    /**
     * 
     * 功能说明: 处理远程终端返回的响应结果
     * 
     * @param ctx 上下文
     * @param cmd 返回结果
     */
    public void processResponseCommand(final ChannelHandlerContext ctx, final RemotingCommand cmd) {
        final ResponseFuture responseFuture = responseTable.get(cmd.getOpaque());
        if (responseFuture != null) {
            getLog().debug(
                "NETTY-COMMON: recive a response opaque[{}] type[{}] result[{}] from channel[{}]",
                cmd.getOpaque(), RequestCode.getName(responseFuture.getCode()),
                ResponseCode.getName(cmd.getCode()), ctx.channel());
            responseFuture.putResponse(cmd);
            responseFuture.release();
            if (responseFuture.getInvokeCallback() != null) {
                this.getCallBackExecutor().submit(new Runnable() {
                    @Override
                    public void run() {
                        try {
                            responseFuture.executeInvokeCallback();
                        } catch (Throwable e) {
                            getLog().error(
                                "NETTY-COMMON: excute async response callback opaque[{}] type[{}] result[{}] from channel[{}] in executor exception:",
                                cmd.getOpaque(), RequestCode.getName(responseFuture.getCode()),
                                ResponseCode.getName(cmd.getCode()), ctx.channel(), e);
                        }
                    }
                });
            }
        } else {
            getLog().debug(
                "NETTY-COMMON: receive a response opaque[{}] result[{}] from channel[{}] but not matched responseFuture remotingCommand[{}]",
                cmd.getOpaque(), ResponseCode.getName(cmd.getCode()), ctx.channel(), cmd);
        }
        responseTable.remove(cmd.getOpaque());
    }
    
    /**
     * 
     * 功能说明:扫描等待结果超时的异步请求
     *
     */
    public void scanTimeoutAsyncResponseTable() {
        Iterator<Entry<Integer, ResponseFuture>> it = responseTable.entrySet().iterator();
        while (it.hasNext()) {
            Entry<Integer, ResponseFuture> next = it.next();
            ResponseFuture responseFuture = next.getValue();
            if (responseFuture.isTimeout()) {
                it.remove();
                try {
                    responseFuture.executeInvokeCallback();
                } catch (Exception e) {
                    getLog().error(
                        "NETTY-COMMON:execute a async response callback method exception:", e);
                } finally {
                    responseFuture.release();
                }
                getLog().warn("NETTY-COMMON: scan a timeout async response{}", responseFuture);
            }
        }
    }
    
    /**
     * 
     * 功能说明:同步调用实现
     * 
     * @param channel 通道
     * @param request 请求
     * @param reponseTimeoutMillis 超时时间
     * @return 响应结果
     * @throws TspRemotingException
     */
    public RemotingCommand invokeSyncImpl(Channel channel, final RemotingCommand request,
        long reponseTimeoutMillis) throws TspRemotingException {
        try {
            final ResponseFuture responseFuture = new ResponseFuture(request.getOpaque(),
                request.getCode(), reponseTimeoutMillis);
            responseTable.put(request.getOpaque(), responseFuture);
            channel.writeAndFlush(request).addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture f) throws Exception {
                    if (f.isSuccess()) {
                        responseFuture.setSendRequestOK(true);
                        return;
                    } else {
                        responseFuture.setSendRequestOK(false);
                    }
                    responseFuture.putResponse(null);
                    responseFuture.setCause(f.cause());
                    responseTable.remove(request.getOpaque());
                }
            });
            try {
                RemotingCommand responseCommand = responseFuture.waitResponse();
                if (null == responseCommand) {
                    if (responseFuture.isSendRequestOK()) {
                        String errorMsg = String.format("send a request timeout[%dms]]",
                            reponseTimeoutMillis);
                        throw new TspRemotingException(errorMsg, NettyExceptionType.REQUEST_TIME_OUT);
                    } else {
                        String errorMsg = String.format("send a request exception: [%s]",
                            responseFuture.getCause().getMessage());
                        throw new TspRemotingException(errorMsg, NettyExceptionType.REQUEST_SEND_FAIL);
                    }
                }
                return responseCommand;
            } catch (InterruptedException e) {
                getLog()
                    .error("NETTY-COMMON: send a sync request waitting response is interrupted.");
                throw new TspRemotingException(
                    "send a sync request waitting response is interrupted.",
                    NettyExceptionType.REQUEST_SEND_FAIL);
            }
        } finally {
            responseTable.remove(request.getOpaque());
        }
    }
    
    public void invokeAsyncImpl(final Channel channel, final RemotingCommand request,
        final long semaphoreTimeoutMillis, final long reponseTimeoutMillis, final INettyInvokeCallback invokeCallback) {
        invokeAsyncImpl(channel, request, semaphoreTimeoutMillis, reponseTimeoutMillis, invokeCallback, null);
    }
    
    /**
     * 
     * 功能说明:异步调用实现
     * 
     * @param channel 通道
     * @param request 请求
     * @param semaphoreTimeoutMillis 获取信号量超时时间
     * @param reponseTimeoutMillis 返回响应结果超时时间
     * @param invokeCallback 回调方法
     */
    public void invokeAsyncImpl(final Channel channel, final RemotingCommand request,
        final long semaphoreTimeoutMillis, final long reponseTimeoutMillis, final INettyInvokeCallback invokeCallback, final INettyInvokeCallback invokeFailCallback) {
        try {
            boolean acquired = semaphoreAsync.tryAcquire(semaphoreTimeoutMillis, TimeUnit.MILLISECONDS);
            if (acquired) {
                SemaphoreReleaseOnlyOnce once = new SemaphoreReleaseOnlyOnce(semaphoreAsync);
                final ResponseFuture responseFuture = new ResponseFuture(request.getOpaque(),
                    request.getCode(), reponseTimeoutMillis, once, invokeCallback);
                responseTable.put(request.getOpaque(), responseFuture);
                channel.writeAndFlush(request).addListener(new ChannelFutureListener() {
                    @Override
                    public void operationComplete(ChannelFuture f) throws Exception {
                        if (f.isSuccess()) {
                            responseFuture.setSendRequestOK(true);
                            return;
                        } else {
                            responseFuture.setSendRequestOK(false);
                            getLog().error(
                                "NETTY-COMMON: send a async request fail to channel[{}] fail that body=[{}] cause:",
                                RemotingUtil.parseChannelRemote(channel), request, f.cause());
                        }
                        responseFuture.release();
                        responseTable.remove(request.getOpaque());
                        
                        try {
                            if(invokeFailCallback != null){
                                invokeFailCallback.operationComplete(request);
                            }
                        } catch (Throwable e) {
                            getLog().error(
                                "NETTY-COMMON: excute async request callback opaque[{}] type[{}] channel[{}] in executor exception:",
                                request.getOpaque(), RequestCode.getName(request.getCode()), channel, e);
                        }
                    }
                });
                return;
            } else {
                getLog().error("NETTY-COMMON: send too many async request or busy system that semaphore status: size[{}] - available[{}]",
                    semaphoreOneway.getQueueLength(), semaphoreOneway.availablePermits());
            }
        } catch (InterruptedException e) {
            getLog().error(
                "NETTY-COMMON: send a async request try acquire a semaphore exception: that waitting is interrupted.");
        }
        
        try {
            if(invokeFailCallback != null){
                invokeFailCallback.operationComplete(request);
            }
        } catch (Throwable e) {
            getLog().error(
                "NETTY-COMMON: excute async request callback opaque[{}] type[{}] channel[{}] in executor exception:",
                request.getOpaque(), RequestCode.getName(request.getCode()), channel, e);
        }
        
    }
    
    /**
     * 
     * 功能说明:oneway调用实现
     * 
     * @param channel 通道
     * @param request 请求
     * @param semaphoreTimeoutMillis 获取信号量超时时间
     */
    public void invokeOnewayImpl(final Channel channel, final RemotingCommand request,
        long semaphoreTimeoutMillis, final INettyInvokeCallback invokeFailCallback) {
        request.markOneway();
        try {
            boolean acquired = semaphoreOneway.tryAcquire(semaphoreTimeoutMillis, TimeUnit.MILLISECONDS);
            if (acquired) {
                final SemaphoreReleaseOnlyOnce once = new SemaphoreReleaseOnlyOnce(semaphoreOneway);
                channel.writeAndFlush(request).addListener(new ChannelFutureListener() {
                    @Override
                    public void operationComplete(ChannelFuture f) throws Exception {
                        once.release();
                        if (!f.isSuccess()) {
                            getLog().error("NETTY-COMMON: send a one way request fail to channel[{}] that body=[{}] cause:",
                                RemotingUtil.parseChannelRemote(channel), request, f.cause());
                            try {
                                if(invokeFailCallback != null){
                                    invokeFailCallback.operationComplete(request);
                                }
                            } catch (Throwable e) {
                                getLog().error(
                                    "NETTY-COMMON: excute oneway request callback opaque[{}] type[{}] channel[{}] in executor exception:",
                                    request.getOpaque(), RequestCode.getName(request.getCode()), channel, e);
                            }
                        }
                    }
                });
                return;
            } else {
                getLog().error("NETTY-COMMON: send too many one way request or busy system that semaphore status: size[{}] - available[{}]",
                    semaphoreOneway.getQueueLength(), semaphoreOneway.availablePermits());
            }
        } catch (InterruptedException e1) {
            getLog().error(
                "NETTY-COMMON: send a one way request try acquire a semaphore exception: that waitting is interrupted.");
        }
        
        try {
            if(invokeFailCallback != null){
                invokeFailCallback.operationComplete(request);
            }
        } catch (Throwable e) {
            getLog().error(
                "NETTY-COMMON: excute oneway request callback opaque[{}] type[{}] channel[{}] in executor exception:",
                request.getOpaque(), RequestCode.getName(request.getCode()), channel, e);
        }
    }
    
    /**
     * 功能说明: 获取注册的各个RPC处理器(远程过程调用协议)Integer(requestcode)
     *
     * @return processorTable 注册的各个RPC处理器(远程过程调用协议)Integer(requestcode)
     */
    public HashMap<Integer, RpcPair<INettyProcessor, ExecutorService>> getProcessorTable() {
        return processorTable;
    }
    
    /**
     * 功能说明: 获取缓存所有对外请求Integer(opaque)
     *
     * @return responseTable 缓存所有对外请求Integer(opaque)
     */
    public ConcurrentHashMap<Integer, ResponseFuture> getResponseTable() {
        return responseTable;
    }
    
    /**
     * 功能说明: 获取通信响应回调处理线程池
     *
     * @return callBackExecutor 通信服务端处理请求应答线程池
     */
    public abstract ExecutorService getCallBackExecutor();
    
    /**
     * 功能说明: 获取通信服务端处理请求线程池
     *
     * @return processsorExecutor 通信服务端处理请求线程池
     */
    public abstract ExecutorService getProcesssorExecutor();
    
    /**
     * 
     * 功能说明:获取日志
     * 
     * @return 日志
     */
    public abstract Logger getLog();
    
}
